Laravel Broadcast——广播系统源码剖析

前言

在现代的 web 应用程序中，WebSockets 被用来实现需要实时、即时更新的接口。当服务器上的数据被更新后，更新信息将通过 WebSocket 连接发送到客户端等待处理。相比于不停地轮询应用程序，WebSocket 是一种更加可靠和高效的选择。

我们先用一个电子商务网站作为例子来概览一下事件广播。当用户在查看自己的订单时，我们不希望他们必须通过刷新页面才能看到状态更新。我们希望一旦有更新时就主动将更新信息广播到客户端。

laravel 的广播系统和队列系统类似，需要两个进程协作，一个是 laravel 的 web 后台系统，另一个是 Socket.IO 服务器系统。具体的流程是页面加载时，网页 js 程序 Laravel Echo 与 Socket.IO 服务器建立连接， laravel 发起通过驱动发布广播，Socket.IO 服务器接受广播内容，对连接的客户端网页推送信息，以达到网页实时更新的目的。

laravel 发起广播的方式有两种，redis 与 pusher。对于 redis 来说，需要支持 Socket.IO 服务器系统，官方推荐 nodejs 为底层的 tlaverdure/laravel-echo-server。对于 pusher 来说，该第三方服务包含了驱动与 Socket.IO 服务器。

本文将会介绍 redis 为驱动的广播源码，由于 laravel-echo-server 是 nodejs 编写，本文也无法介绍 Socket.IO 方面的内容。

广播系统服务的启动

和其他服务类似，广播系统服务的注册实质上就是对 Ioc 容器注册门面类，广播系统的门面类是 BroadcastManager:

  1. class BroadcastServiceProvider extends ServiceProvider
  2. {
  3.     public function register()
  4.     {
  5.         $this->app->singleton(BroadcastManager::class, function ($app) {
  6.             return new BroadcastManager($app);
  7.         });
  9.         $this->app->singleton(BroadcasterContract::class, function ($app) {
  10.             return $app->make(BroadcastManager::class)->connection();
  11.         });
  13.         $this->app->alias(
  14.             BroadcastManager::class, BroadcastingFactory::class
  15.         );
  16.     }
  17. }

除了注册 BroadcastManager，BroadcastServiceProvider 还进行了广播驱动的启动：

  1. public function connection($driver = null)
  2. {
  3.     return $this->driver($driver);
  4. }
  6. public function driver($name = null)
  7. {
  8.     $name = $name ?: $this->getDefaultDriver();
  10.     return $this->drivers[$name] = $this->get($name);
  11. }
  13. protected function get($name)
  14. {
  15.     return isset($this->drivers[$name]) ? $this->drivers[$name] : $this->resolve($name);
  16. }
  18. protected function resolve($name)
  19. {
  20.     $config = $this->getConfig($name);
  22.     if (is_null($config)) {
  23.         throw new InvalidArgumentException("Broadcaster [{$name}] is not defined.");
  24.     }
  26.     if (isset($this->customCreators[$config['driver']])) {
  27.         return $this->callCustomCreator($config);
  28.     }
  30.     $driverMethod = 'create'.ucfirst($config['driver']).'Driver';
  32.     if (! method_exists($this, $driverMethod)) {
  33.         throw new InvalidArgumentException("Driver [{$config['driver']}] is not supported.");
  34.     }
  36.     return $this->{$driverMethod}($config);
  37. }
  39. protected function createRedisDriver(array $config)
  40. {
  41.     return new RedisBroadcaster(
  42.         $this->app->make('redis'), Arr::get($config, 'connection')
  43.     );
  44. }

广播信息的发布

广播信息的发布与事件的发布大致相同，要告知 Laravel 一个给定的事件是广播类型，只需在事件类中实现 Illuminate\Contracts\Broadcasting\ShouldBroadcast 接口即可。该接口已经被导入到所有由框架生成的事件类中，所以可以很方便地将它添加到自己的事件中。

ShouldBroadcast 接口要求你实现一个方法：broadcastOn. broadcastOn 方法返回一个频道或一个频道数组，事件会被广播到这些频道。频道必须是 Channel、PrivateChannel 或 PresenceChannel 的实例。Channel 实例表示任何用户都可以订阅的公开频道，而 PrivateChannels 和 PresenceChannels 则表示需要 频道授权 的私有频道：

  1. class ServerCreated implements ShouldBroadcast
  2. {
  3.     use SerializesModels;
  5.     public $user;
  7.     //默认情况下，每一个广播事件都被添加到默认的队列上，默认的队列连接在 queue.php 配置文件中指定。可以通过在事件类中定义一个 broadcastQueue 属性来自定义广播器使用的队列。该属性用于指定广播使用的队列名称：
  8.     public $broadcastQueue = 'your-queue-name';
  10.     public function __construct(User $user)
  11.     {
  12.         $this->user = $user;
  13.     }
  15.     public function broadcastOn()
  16.     {
  17.         return new PrivateChannel('user.'.$this->user->id);
  18.     }
  20.     //Laravel 默认会使用事件的类名作为广播名称来广播事件，自定义：
  21.     public function broadcastAs()
  22.     {
  23.         return 'server.created';
  24.     }
  26.     //想更细粒度地控制广播数据:
  27.     public function broadcastWith()
  28.     {
  29.         return ['id' => $this->user->id];
  30.     }
  32.     //有时，想在给定条件为 true ，才广播事件:
  33.     public function broadcastWhen()
  34.     {
  35.         return $this->value > 100;
  36.     }
  37. }

然后，只需要像平时那样触发事件。一旦事件被触发，一个队列任务会自动广播事件到你指定的广播驱动器上。

当一个事件被广播时，它所有的 public 属性会自动被序列化为广播数据，这允许你在你的 JavaScript 应用中访问事件的公有数据。因此，举个例子，如果你的事件有一个公有的 $user 属性，它包含了一个 Elouqent 模型，那么事件的广播数据会是：

  1. {
  2.     "user": {
  3.         "id": 1,
  4.         "name": "Patrick Stewart"
  5.         ...
  6.     }
  7. }

广播发布的源码

广播的发布与事件的触发是一体的，具体的流程我们已经在 event 的源码中介绍清楚了，现在我们来看唯一的不同：

  1. public function dispatch($event, $payload = [], $halt = false)
  2.     {
  3.         list($event, $payload) = $this->parseEventAndPayload(
  4.             $event, $payload
  5.         );
  7.         if ($this->shouldBroadcast($payload)) {
  8.             $this->broadcastEvent($payload[0]);
  9.         }
  11.         ...
  12.     }
  14.     protected function shouldBroadcast(array $payload)
  15.     {
  16.         return isset($payload[0]) && $payload[0] instanceof ShouldBroadcast;
  17.     }
  19.     protected function broadcastEvent($event)
  20.     {
  21.         $this->container->make(BroadcastFactory::class)->queue($event);
  22.     }

可见，关键之处在于 BroadcastManager 的 quene 方法：

  1. public function queue($event)
  2. {
  3.     $connection = $event instanceof ShouldBroadcastNow ? 'sync' : null;
  5.     if (is_null($connection) && isset($event->connection)) {
  6.         $connection = $event->connection;
  7.     }
  9.     $queue = null;
  11.     if (isset($event->broadcastQueue)) {
  12.         $queue = $event->broadcastQueue;
  13.     } elseif (isset($event->queue)) {
  14.         $queue = $event->queue;
  15.     }
  17.     $this->app->make('queue')->connection($connection)->pushOn(
  18.         $queue, new BroadcastEvent(clone $event)
  19.     );
  20. }

可见，quene 方法将广播事件包装为事件类，并且通过队列发布，我们接下来看这个事件类的处理：

  1. class BroadcastEvent implements ShouldQueue
  2. {
  3.     public function handle(Broadcaster $broadcaster)
  4.     {
  5.         $name = method_exists($this->event, 'broadcastAs')
  6.                 ? $this->event->broadcastAs() : get_class($this->event);
  8.         $broadcaster->broadcast(
  9.             array_wrap($this->event->broadcastOn()), $name,
  10.             $this->getPayloadFromEvent($this->event)
  11.         );
  12.     }
  14.     protected function getPayloadFromEvent($event)
  15.     {
  16.         if (method_exists($event, 'broadcastWith')) {
  17.             return array_merge(
  18.                 $event->broadcastWith(), ['socket' => data_get($event, 'socket')]
  19.             );
  20.         }
  22.         $payload = [];
  24.         foreach ((new ReflectionClass($event))->getProperties(ReflectionProperty::IS_PUBLIC) as $property) {
  25.             $payload[$property->getName()] = $this->formatProperty($property->getValue($event));
  26.         }
  28.         return $payload;
  29.     }
  31.     protected function formatProperty($value)
  32.     {
  33.         if ($value instanceof Arrayable) {
  34.             return $value->toArray();
  35.         }
  37.         return $value;
  38.     }
  39. }

可见该事件主要调用 broadcaster 的 broadcast 方法，我们这里讲 redis 的发布：

  1. class RedisBroadcaster extends Broadcaster
  2. {
  3.     public function broadcast(array $channels, $event, array $payload = [])
  4.     {
  5.         $connection = $this->redis->connection($this->connection);
  7.         $payload = json_encode([
  8.             'event' => $event,
  9.             'data' => $payload,
  10.             'socket' => Arr::pull($payload, 'socket'),
  11.         ]);
  13.         foreach ($this->formatChannels($channels) as $channel) {
  14.             $connection->publish($channel, $payload);
  15.         }
  16.     }
  17. }
  19. protected function formatChannels(array $channels)
  20. {
  21.     return array_map(function ($channel) {
  22.         return (string) $channel;
  23.     }, $channels);
  24. }

broadcast 方法运用了 redis 的 publish 方法，对 redis 进行了频道的信息发布。

频道授权

对于私有频道，用户只有被授权后才能监听。实现过程是用户向 Laravel 应用程序发起一个携带频道名称的 HTTP 请求，应用程序判断该用户是否能够监听该频道。在使用 Laravel Echo 时，上述 HTTP 请求会被自动发送；尽管如此，仍然需要定义适当的路由来响应这些请求。

定义授权路由

我们可以在 Laravel 里很容易地定义路由来响应频道授权请求。

  1. Broadcast::routes();

Broadcast::routes 方法会自动把它的路由放进 web 中间件组中；另外，如果你想对一些属性自定义，可以向该方法传递一个包含路由属性的数组

  1. Broadcast::routes($attributes);

定义授权回调

接下来，我们需要定义真正用于处理频道授权的逻辑。这是在 routes/channels.php 文件中完成。在该文件中，你可以用 Broadcast::channel 方法来注册频道授权回调函数：

  1. Broadcast::channel('order.{orderId}', function ($user, $orderId) {
  2.     return $user->id === Order::findOrNew($orderId)->user_id;
  3. });

channel 方法接收两个参数：频道名称和一个回调函数，该回调通过返回 true 或 false 来表示用户是否被授权监听该频道。

所有的授权回调接收当前被认证的用户作为第一个参数，任何额外的通配符参数作为后续参数。在本例中，我们使用 {orderId} 占位符来表示频道名称的「ID」部分是通配符。

授权回调模型绑定

就像 HTTP 路由一样，频道路由也可以利用显式或隐式 路由模型绑定。例如，相比于接收一个字符串或数字类型的 order ID，你也可以请求一个真正的 Order 模型实例:

  1. Broadcast::channel('order.{order}', function ($user, Order $order) {
  2.     return $user->id === $order->user_id;
  3. });

频道授权源码分析

授权路由

  1. class BroadcastManager implements FactoryContract
  2. {
  3.     public function routes(array $attributes = null)
  4.     {
  5.         if ($this->app->routesAreCached()) {
  6.             return;
  7.         }
  9.         $attributes = $attributes ?: ['middleware' => ['web']];
  11.         $this->app['router']->group($attributes, function ($router) {
  12.             $router->post('/broadcasting/auth', BroadcastController::class.'@authenticate');
  13.         });
  14.     }
  15. }

频道专门有 Controller 来处理授权服务：

  1. class BroadcastController extends Controller
  2. {
  3.     public function authenticate(Request $request)
  4.     {
  5.         return Broadcast::auth($request);
  6.     }
  7. }

当 Socket Io 服务器对 javascript 程序推送数据的时候，首先会经过该 controller 进行授权验证：

  1. public function auth($request)
  2. {
  3.     if (Str::startsWith($request->channel_name, ['private-', 'presence-']) &&
  4.         ! $request->user()) {
  5.         throw new HttpException(403);
  6.     }
  8.     $channelName = Str::startsWith($request->channel_name, 'private-')
  9.                         ? Str::replaceFirst('private-', '', $request->channel_name)
  10.                         : Str::replaceFirst('presence-', '', $request->channel_name);
  12.     return parent::verifyUserCanAccessChannel(
  13.         $request, $channelName
  14.     );
  15. }

verifyUserCanAccessChannel 根据频道与其绑定的闭包函数来验证该频道是否可以通过授权：

  1. protected function verifyUserCanAccessChannel($request, $channel)
  2. {
  3.     foreach ($this->channels as $pattern => $callback) {
  4.         if (! Str::is(preg_replace('/\{(.*?)\}/', '*', $pattern), $channel)) {
  5.             continue;
  6.         }
  8.         $parameters = $this->extractAuthParameters($pattern, $channel, $callback);
  10.         if ($result = $callback($request->user(), ...$parameters)) {
  11.             return $this->validAuthenticationResponse($request, $result);
  12.         }
  13.     }
  15.     throw new HttpException(403);
  16. }

由于频道的命名经常带有 userid 等参数，因此判断频道之前首先要把 channels 中的频道名转为通配符 *，例如 order.{userid} 转为 order.*，之后进行正则匹配。

extractAuthParameters 用于提取频道的闭包函数的参数，合并 $request->user() 之后调用闭包函数。

  1. protected function extractAuthParameters($pattern, $channel, $callback)
  2. {
  3.     $callbackParameters = (new ReflectionFunction($callback))->getParameters();
  5.     return collect($this->extractChannelKeys($pattern, $channel))->reject(function ($value, $key) {
  6.         return is_numeric($key);
  7.     })->map(function ($value, $key) use ($callbackParameters) {
  8.         return $this->resolveBinding($key, $value, $callbackParameters);
  9.     })->values()->all();
  10. }
  12. protected function extractChannelKeys($pattern, $channel)
  13. {
  14.     preg_match('/^'.preg_replace('/\{(.*?)\}/', '(?<$1>[^\.]+)', $pattern).'/', $channel, $keys);
  16.     return $keys;
  17. }
  19. public function validAuthenticationResponse($request, $result)
  20. {
  21.     if (is_bool($result)) {
  22.         return json_encode($result);
  23.     }
  25.     return json_encode(['channel_data' => [
  26.         'user_id' => $request->user()->getKey(),
  27.         'user_info' => $result,
  28.     ]]);
  29. }

extractChannelKeys 用于将 order.{userid} 与 order.23 中 userid 和 23 建立 key、value 关联。如果 userid 是 User 的主键，resolveBinding 还可以为其自动进行路由模型绑定。